package chen.bupt.util;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import chen.bupt.constant.Constants;

public class HDFSFileUtils {
	private static final String default_prefix = "part";
	private static final String default_hdfs = "hdfs://10.1.1.26:31054";
	@SuppressWarnings("unchecked")
	public static boolean isInputFile(Context context){
		return isInputFile(context, default_prefix);
	}
	
	@SuppressWarnings("unchecked")
	public static boolean isInputFile(Context context, String prefix){
		FileSplit fs = (FileSplit) context.getInputSplit();
		String fileName = fs.getPath().getName();
		if (fileName == null || "".equals(fileName)) {
			return false;
		}
		if (!fileName.startsWith(prefix)) {
			return false;
		}
		return true;
	}
	
	public static void deleteFile(Path path, Configuration conf) throws Exception{
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(path)) {
			fs.delete(path, true);
		}
	}

	/**
	 * read file from HDFS
	 * @param HDFSPath
	 * 					file path in HDFS
	 * @return
	 */
	public static List<String> readLines(String HDFSPath, String HDFSURL){
		try {
			Configuration conf = new Configuration();
			conf.set("fs.default.name", HDFSURL);
			FileSystem fs = FileSystem.get(conf);
			FSDataInputStream fis = fs.open(new Path(HDFSPath));
			BufferedReader br = new BufferedReader(new InputStreamReader(fis));
			List<String> list = new ArrayList<String>();
			String str = br.readLine();
			while(str!=null){
				list.add(str);
				str = br.readLine();
			}
			br.close();
			return list;
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
	
	public static List<String> readLines(String HDFSPath){
		return readLines(HDFSPath, default_hdfs);
	}
	
	public static void write2HDFS(List lines, String HDFSPath, String HDFSURL){
		try {
			Configuration conf = new Configuration();
			conf.set("fs.default.name", HDFSURL);
			FileSystem fs = FileSystem.get(conf);
			FSDataOutputStream dos = fs.create(new Path(HDFSPath));
			for(Object s:lines){
				dos.writeBytes(s.toString()+"\n");
			}
			dos.flush();
			dos.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void write2HDFS(List lines, String HDFSPath){
		write2HDFS(lines, HDFSPath, default_hdfs);
	}
}
